	/***********************************************************************
	 * Mirrors most methods of Filter Framework for a second Input Read Port and Input Filter
	 * Merging data is very tricky because one does not want to mix data from different input sources
	 * Each input source is stored into a queue until an entire frame from the input source arrives
	 * Then, one must determine which frame to send, based on the time stamp.
	 * The entire frame is logically sent at a single time, but behind the scenes, 
	 * it simply calls several methods repeatedly to send increments at a time. 
	 ***********************************************************************/

import java.io.IOException;
import java.io.PipedInputStream;
import java.util.LinkedList;

public class MergeFilter extends FrameFilterFramework{

	private PipedInputStream InputReadPort2 = new PipedInputStream();

	private FilterFramework InputFilter2;

	//holds frames of data to to send
	private LinkedList<Frame> frameQueue = new LinkedList<Frame>();
	private LinkedList<Frame> frameQueue2 = new LinkedList<Frame>();
	
	//records state of inputs
	boolean inputAlive = true;
	boolean input2Alive = true;

	int IdLength = 4;				// This is the length of IDs in the byte stream
	int bytesread = 0;				// This is the number of bytes read from the stream
	int MeasurementLength = 8;		// This is the length of all measurements (including time) in bytes

	public void run()
	{	
		// Next we write a message to the terminal to let the world know we are alive...

		System.out.print( "\n" + this.getName() + "::Merge Reading \n");
		
		/*************************************************************
		 *	main loop while either input is alive
		 **************************************************************/
		while (
				inputAlive ||
				input2Alive
				){

			try{
				while (true){
					if (inputAlive){
						processInput(Input.FilterPort1);
					}
					if (input2Alive){
						processInput(Input.FilterPort2);
					}
				}
			}
			catch(EndOfStreamException e){
				this.ClosePorts();
				
				//break;
			}
		}
		
		/*************************************************************
		 *	Cleanup for after the inputs are closed
		 **************************************************************/
		while(
				frameQueue.size()!=0 || 
				frameQueue2.size()!=0
				){
			//if only frameQueue has members remaining
			if(
					frameQueue2.size()==0 && 
					frameQueue.size()!=0
					){
				while (frameQueue.size()!=0){
					this.writeFrameToOutputFilterPort(frameQueue.getFirst());
					frameQueue.remove();
				}
			}
			//if only frameQueue2 has members remaining
			else if(
					frameQueue.size()==0 && 
					frameQueue2.size()!=0
					){
				while (frameQueue2.size()!=0){
					this.writeFrameToOutputFilterPort(frameQueue2.getFirst());
					frameQueue2.remove();
				}
			}
			//both have members remaining
			else if(
					frameQueue.size()>0&&
					frameQueue2.size()>0){
				while(
						frameQueue.size()>0 &&
						frameQueue2.size()>0){
					if(frameQueue.getFirst().getTime()<frameQueue2.getFirst().getTime()){
						this.writeFrameToOutputFilterPort(frameQueue.getFirst());
						frameQueue.remove();
					}
					else{
						this.writeFrameToOutputFilterPort(frameQueue2.getFirst());
						frameQueue2.remove();
					}
				}
			}
		}
		/*************************************************************
		 *	Finished and closing port
		 **************************************************************/
		
		try {
			OutputWritePort.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		//System.out.println("frameQueue size: "+frameQueue.size()+"; frameQueue2 size: "+frameQueue2.size());
		System.out.print( "\n" + this.getName() + "::Merge Exiting; bytes read: " + bytesread+"\n");
	} // run

	private void processInput(Input input) throws EndOfStreamException {
		long measurement;				// This is the word used to store all measurements - conversions are illustrated.
		int id;							// This is the measurement id		

			/*************************************************************
			 *	Read ids
			 **************************************************************/

			id = readID(input);

			/*************************************************************
			 *	Read measurements
			 **************************************************************/

			measurement = readMeasurement(id, input);

			/*************************************************************
			 *	Add to Frame and possibly send the frame, as appropriate
			 **************************************************************/

			addToFrame(id, measurement, input);

			//handles case where both inputs are alive
			if(inputAlive && input2Alive){
				//if both queues have full frames in the first frames of the queue ...
				
				//System.out.println("frameQueue size: "+frameQueue.size()+"; frameQueue2 size: "+frameQueue2.size());
				if(
						frameQueue.size()>0 &&
						frameQueue2.size()>0
						){
					
					Frame frame = frameQueue.getFirst();
					Frame frame2 = frameQueue2.getFirst();
					if (
							frame.isFullFrame()  && 
							frame2.isFullFrame()
						){
						//
						//compare the time stamps of both to find the one with the earliest timestamp
						//then write the entire frame to the output filter port and remove it from the queue
						if(frame.getTime()<frame2.getTime()){
							this.writeFrameToOutputFilterPort(frame);
							frameQueue.removeFirst();
						}
						else{
							this.writeFrameToOutputFilterPort(frame2);
							frameQueue2.removeFirst();
						}
					}
				}
			}
			//handles only the first input alive
			else if (inputAlive){
				Frame frame = frameQueue.getFirst();
				if (
						frame != null &&
						frame.isFullFrame()
					){
					this.writeFrameToOutputFilterPort(frame);
					frameQueue.removeFirst();
				}
			}
			//handles only the second input alive
			else {
				Frame frame = frameQueue2.getFirst();
				if (
						frame != null &&
						frame.isFullFrame()
						){
					this.writeFrameToOutputFilterPort(frame);
					frameQueue2.removeFirst();
				}
			}

	}

	private void addToFrame(int id, long measurement, Input input) {

		//for inputQueue1
		if (input == Input.FilterPort1){

			//checks all frames in the queue to see if they are missing the measurement
			//if the frame is missing the measurement, it is added to the frame
			boolean addedMeasurement = false;
			for (int i = 0; i<frameQueue.size(); i++){
				addedMeasurement = frameQueue.get(i).addMeasurement(id, measurement);
				if (addedMeasurement == true){
					break;
				}
			}
			//if there were no frames missing that measurement
			if (addedMeasurement == false){
				Frame frame = new Frame();
				frame.addMeasurement(id, measurement);
				frameQueue.add(frame);
			}

		}

		//same for inputQueue2
		else{
			//checks all frames in the queue to see if they are missing the measurement
			//if the frame is missing the measurement, it is added to the frame
			boolean addedMeasurement = false;
			for (int i = 0; i<frameQueue2.size(); i++){
				addedMeasurement = frameQueue2.get(i).addMeasurement(id, measurement);
				if (addedMeasurement == true){
					break;
				}
			}
			//if there were no frames missing that measurement
			if (addedMeasurement == false){
				Frame frame = new Frame();
				frame.addMeasurement(id, measurement);
				frameQueue2.add(frame);
			}
		}

	}

	void Connect2( FilterFramework Filter )
	{
		try
		{
			// Connect this filter's input to the upstream pipe's output stream

			//TODO:OutputWritePort now protected
			InputReadPort2.connect( Filter.OutputWritePort );
			InputFilter2 = Filter;

		} // try

		catch( Exception Error )
		{
			System.out.println( "\n" + this.getName() + " FilterFramework error connecting::"+ Error );

		} // catch

	} // Connect2

	byte ReadFilterInputPort2() throws EndOfStreamException
	{
		byte datum = 0;

		try
		{
			while (InputReadPort2.available()==0 )
			{
				if (EndOfInputStream2())
				{
					throw new EndOfStreamException("End of input stream 2 reached");

				} //if

				sleep(250);

			} // while

		} // try

		catch( EndOfStreamException Error )
		{
			throw Error;

		} // catch

		catch( Exception Error )
		{
			System.out.println( "\n" + this.getName() + " Error in read port 2 wait loop::" + Error );

		} // catch

		/***********************************************************************
		 * If at least one byte of data is available on the input
		 * pipe we can read it. We read and write one byte to and from ports.
		 ***********************************************************************/

		try
		{
			datum = (byte)InputReadPort2.read();
			return datum;

		} // try

		catch( Exception Error )
		{
			System.out.println( "\n" + this.getName() + " Pipe 2 read error::" + Error );
			return datum;

		} // catch

	} // ReadFilterPort

	private boolean EndOfInputStream2()
	{
		if (InputFilter2.isAlive())
		{
			return false;

		} else {

			return true;

		} // if

	} // EndOfInputStream

	//TODO: Note that ClosePorts is overidden, only closes the input ports
	void ClosePorts()
	{
		try
		{
			//TODO: Note that InputFilter had to be changed to protected instead of private
			//TODO: Note that the state of the inputs is recorded
			if (
					inputAlive &&
					!InputFilter.isAlive()
					){
				//TODO Note that InputReadPort is changed to protected instead of private
				InputReadPort.close();
				inputAlive = false;
				System.out.println(this.getName()+"::Merge first input filter closed.");
			}
			if (
					input2Alive &&
					!InputFilter2.isAlive()
					){
				InputReadPort2.close();
				input2Alive = false;
				System.out.println(this.getName()+"::Merge second input filter closed.");
			}
			
		}
		catch( Exception Error )
		{
			System.out.println( "\n" + this.getName() + " ClosePorts error::" + Error );

		} // catch

	} // ClosePorts

	private int readID(Input input) throws EndOfStreamException{
		//Reads id from the appropriate input source
		byte databyte;
		int id = 0;
		for (int i=0; i<IdLength; i++ )
		{
			if (input == Input.FilterPort1){
				databyte = ReadFilterInputPort();
			}
			else{
				databyte = ReadFilterInputPort2();	// This is where we read the byte from the stream...
			}

			id = id | (databyte & 0xFF);		// We append the byte on to ID...

			if (i != IdLength-1)				// If this is not the last byte, then slide the
			{									// previously appended byte to the left by one byte
				id = id << 8;					// to make room for the next byte we append to the ID

			} // if

			bytesread++;						// Increment the byte count

		} // for
		return id;
	}

	private long readMeasurement(int id, Input input) throws EndOfStreamException {
		//Reads measurement from the appropriate input source
		long measurement = 0;
		byte databyte = 0;

		for (int i=0; i<MeasurementLength; i++ )
		{
			if (input == Input.FilterPort1){
				databyte = ReadFilterInputPort();
			}
			else {
				databyte = ReadFilterInputPort2();
			}

			measurement = measurement | (databyte & 0xFF);	// We append the byte on to measurement...

			if (i != MeasurementLength-1)					// If this is not the last byte, then slide the
			{												// previously appended byte to the left by one byte
				measurement = measurement << 8;				// to make room for the next byte we append to the
				// measurement
			} // if

			bytesread++;									// Increment the byte count

		} // if
		return measurement;
	}

	enum Input{
		FilterPort1,
		FilterPort2;
	}
}
